package com.chiancloud.shield.server.services.curator;

import com.chiancloud.shield.core.registry.info.ServiceInstanceData;
import com.chiancloud.shield.server.services.registry.RegistryDiscoveryContext;
import com.chiancloud.shield.server.services.registry.RegistryServiceConstants;
import com.chiancloud.shield.server.services.registry.ShieldRegistryService;
import com.google.common.base.Preconditions;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.strategies.RandomStrategy;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;

/**
 * Created by wanghui on 14-8-20.
 */
public class CuratorHelper extends Configured {

    private final CuratorFramework curator;
    private final String connectionString;

    public CuratorHelper(Configuration conf, String connectionString) {
        super(conf);
        this.connectionString = connectionString;
        curator = createCurator(this.connectionString);
    }
    public CuratorFramework getCurator() {
        return curator;
    }

    public String getConnectionString() {
        return connectionString;
    }

    public RetryPolicy createRetryPolicy() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        return retryPolicy;
    }
    private CuratorFramework createCurator(String connectionString) {
        CuratorFramework curator =
                CuratorFrameworkFactory.newClient(connectionString,
                        createRetryPolicy());
        return curator;
    }


    public CuratorService createCuratorClientService() {
        CuratorService curatorService =
                new CuratorService("Curator ", curator, connectionString);
        return curatorService;
    }


    public ServiceDiscoveryBuilder<ServiceInstanceData> createDiscoveryBuilder() {
        ServiceDiscoveryBuilder<ServiceInstanceData> discoveryBuilder =
                ServiceDiscoveryBuilder.builder(ServiceInstanceData.class);
        discoveryBuilder.client(curator);
        return discoveryBuilder;
    }

    public ShieldRegistryService createRegistryBinderService(
            String basePath,
            ServiceDiscoveryBuilder<ServiceInstanceData> discoveryBuilder) {
        discoveryBuilder.basePath(basePath);
        return new ShieldRegistryService(curator,
                basePath,
                discoveryBuilder.build());
    }


    public ShieldRegistryService createRegistryBinderService(
            String basePath) {
        ServiceDiscoveryBuilder<ServiceInstanceData> discoveryBuilder =
                createDiscoveryBuilder();
        //registry will start curator as well as the binder, in the correct order
        ShieldRegistryService registryBinderService =
                createRegistryBinderService(basePath, discoveryBuilder);
        return registryBinderService;
    }

    public RegistryDiscoveryContext createDiscoveryContext(
            ServiceDiscovery<ServiceInstanceData> discovery) {
        Preconditions.checkNotNull(discovery);
        return new RegistryDiscoveryContext(discovery,
                new RandomStrategy<ServiceInstanceData>(),
                RegistryServiceConstants.INSTANCE_REFRESH_MS,
                ServiceInstanceData.class);

    }
}
